Single-run lock per ingestion job - #93
Open
koenvo wants to merge 4 commits into
Open
Conversation
Enforce (best-effort) that one (source, dataset_type, selector) job never runs in two processes at once -- a framework invariant, not a config knob, since concurrent same-job execution is undefined. Plan: a Postgres session-level advisory lock in the metadata store, held on a dedicated connection so it is released the instant the process ends; no-op on stores without one. Taken in IngestionJob.execute before the RUNNING summary. Design: docs/design/single-run-lock.md Tests: ingestify/tests/test_run_lock.py specify DatasetRepository.acquire_run_lock (currently red: method not implemented yet). Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
koenvo
force-pushed
the
feat/single-run-lock
branch
from
August 7, 2026 08:41
2abb0f1 to
1bcd65f
Compare
- DatasetRepository.acquire_run_lock: Postgres pg_try_advisory_lock, MySQL GET_LOCK, on a dedicated AUTOCOMMIT connection held for the run (released the instant the connection/process ends). Concrete no-op default for stores without a session lock. - DatasetStore delegates it; IngestionJob.execute takes the lock before the RUNNING summary and, if another run holds it, records a SKIPPED summary and returns without running -- releasing the lock in a finally around the existing body. - test_run_lock.py: repository exclusivity/no-op unit tests plus a two-instance integration test (two engines over one DB; instance 1 holds the lock while blocked, instance 2 must skip -- its source is never invoked). Runs on Postgres/MySQL. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
koenvo
changed the title
The threaded version hung on MySQL/Postgres: the background thread left an idle-in-transaction connection its scoped_session could not clean from the main thread, so cleanup's DROP TABLE waited on it forever. Instead hold the run lock directly on a second store connection (exactly what a second instance does), run the other instance -> it must skip, then release and run again -> it runs. No threads. Cleanup now disposes every pool before DROP. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
A run that stops before finishing the whole plan was recorded as FINISHED (StopProcessing) or left stuck in RUNNING (Ctrl-C / SIGTERM, which are BaseExceptions that escaped every `except Exception`). Neither is an error, but neither is a real completion. - New IngestionJobState.ABORTED + set_aborted(); partial results are saved and the next run resumes. - StopProcessing now marks the summary ABORTED (sync and async paths). - KeyboardInterrupt / SystemExit are caught in both work paths -> ABORTED + persisted, instead of a zombie RUNNING row. (A SIGTERM handler that raises KeyboardInterrupt makes Cloud Run task-timeout shutdown flow the same path; the run lock's finally frees the lock on unwind.) - Tests: StopProcessing -> ABORTED and Ctrl-C -> ABORTED (sync); async StopProcessing assertion updated FINISHED -> ABORTED. Claude-Session: https://claude.ai/code/session_01B5EfLJqoafjW1FhvkxGSmg
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
In lots of cases ingestify is scheduled using a cron. This works fine when jobs finish before the next starts. In case an api is slow this might not be the case and things get messy.
This PR tries to solve this by using locks per IngestionJob. Different IngestionJobs can run in parallel (from different sources for example). But the same job from the same Source cannot run in parallel.